#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import json
import time
import yaml
import logging
import argparse
import re
from datetime import datetime
import pymysql
from elasticsearch import Elasticsearch, helpers
from pymysql.cursors import DictCursor
import sys

try:
    import sqlglot
    from sqlglot import parse_one, exp
    HAS_SQLGLOT = True
except ImportError:
    HAS_SQLGLOT = False
    logging.warning("未安装SQLGlot库，将使用简单的字符串解析。建议安装SQLGlot以获得更好的SQL解析支持: pip install sqlglot")

# 创建数据和日志目录
DATA_DIR = os.environ.get('DATA_DIR', './data')
LOGS_DIR = os.environ.get('LOGS_DIR', './logs')

os.makedirs(DATA_DIR, exist_ok=True)
os.makedirs(LOGS_DIR, exist_ok=True)

# 默认日志格式
DEFAULT_LOG_FORMAT = '%(asctime)s - [%(process)d:%(thread)d] - %(pathname)s#%(lineno)d - %(name)s - %(levelname)s - %(message)s'

# 配置日志
def setup_logging(log_level='DEBUG', log_format=None):
    """设置日志级别和格式"""
    # 命令行参数指定的级别最优先
    level = getattr(logging, log_level.upper(), logging.DEBUG)
    
    # 如果未指定格式，使用默认格式
    if log_format is None:
        # 检查全局变量中是否有FORMAT定义
        log_format = globals().get('FORMAT', DEFAULT_LOG_FORMAT)
    
    # 检查根日志记录器是否已经有处理器配置
    # 如果已经有配置，则尊重已有配置，但是覆盖日志级别
    if len(logging.root.handlers) > 0:
        # 使用临时日志器避免在日志系统未完全配置前使用 logging.info
        temp_logger = logging.getLogger('setup')
        temp_logger.setLevel(level)
        temp_logger.info(f"检测到已有日志配置，使用已有配置，但日志级别设置为: {log_level}")
        # 覆盖全局日志级别
        logging.root.setLevel(level)
    else:
        # 没有已有配置，创建新配置
        logging.basicConfig(
            level=level,
            format=log_format,
            handlers=[
                logging.FileHandler(os.path.join(LOGS_DIR, 'mysql_to_es_sync.log')),
                logging.StreamHandler()
            ]
        )
    
    # 返回的logger对象也要设置为指定级别
    logger = logging.getLogger('mysql_to_es_sync')
    logger.setLevel(level)
    return logger

# 默认日志配置
logger = logging.getLogger('mysql_to_es_sync')

class MySQLToESSync:
    """MySQL到Elasticsearch的数据同步工具"""
    
    def __init__(self, config_file):
        """初始化同步器"""
        logger.info(f"初始化同步器，配置文件: {config_file}")
        self.config_file = config_file
        self.load_config()
        self.sync_state_file = os.path.join(DATA_DIR, 'sync_state.json')
        self.sync_state = self.load_sync_state()
        self.mysql_conn = None
        self.es_client = None
        
    def load_config(self):
        """加载配置文件"""
        logger.debug(f"加载配置文件: {self.config_file}")
        with open(self.config_file, 'r', encoding='utf-8') as f:
            if self.config_file.endswith('.yaml') or self.config_file.endswith('.yml'):
                self.config = yaml.safe_load(f)
            else:
                # 假设是python格式的配置
                config_content = f.read()
                local_vars = {}
                exec(config_content, globals(), local_vars)
                self.config = local_vars
                
        # 提取关键配置
        self.mysql_config = self.config.get('CONNECTION', {})
        self.es_nodes = self.config.get('NODES', [])
        self.bulk_size = self.config.get('BULK_SIZE', 1000)
        self.tasks = self.config.get('TASKS', [])
        
        logger.info(f"配置加载完成，任务数量: {len(self.tasks)}")
        
    def load_sync_state(self):
        """加载同步状态"""
        if os.path.exists(self.sync_state_file):
            logger.debug(f"从文件加载同步状态: {self.sync_state_file}")
            try:
                with open(self.sync_state_file, 'r') as f:
                    return json.load(f)
            except Exception as e:
                logger.error(f"加载同步状态失败: {e}")
                return {}
        else:
            logger.debug("未找到同步状态文件，使用空状态")
            return {}
    
    def save_sync_state(self):
        """保存同步状态"""
        logger.debug(f"保存同步状态到文件: {self.sync_state_file}")
        with open(self.sync_state_file, 'w') as f:
            json.dump(self.sync_state, f)
    
    def connect_mysql(self):
        """连接MySQL数据库"""
        logger.info("连接MySQL数据库")
        try:
            # MySQL 5.7兼容连接参数
            self.mysql_conn = pymysql.connect(
                host=self.mysql_config.get('host', 'localhost'),
                port=self.mysql_config.get('port', 3306),
                user=self.mysql_config.get('user', 'root'),
                password=self.mysql_config.get('passwd', ''),
                connect_timeout=self.mysql_config.get('connect_timeout', 60),
                read_timeout=self.mysql_config.get('read_timeout', 300),
                write_timeout=self.mysql_config.get('write_timeout', 300),
                charset='utf8mb4',
                cursorclass=DictCursor,
                autocommit=True  # 添加自动提交配置
            )
            logger.info("MySQL连接成功")
            return True
        except Exception as e:
            logger.error(f"MySQL连接失败: {e}")
            return False
    
    def connect_es(self):
        """连接Elasticsearch"""
        logger.info(f"连接Elasticsearch: {self.es_nodes}")
        try:
            # ES 7.x兼容配置
            self.es_client = Elasticsearch(
                self.es_nodes,
                # 针对ES 7.7的配置参数
                timeout=300,
                max_retries=3,
                retry_on_timeout=True
            )
            if self.es_client.ping():
                logger.info("Elasticsearch连接成功")
                # 获取ES版本号
                es_info = self.es_client.info()
                es_version = es_info.get('version', {}).get('number', '未知')
                logger.info(f"Elasticsearch版本: {es_version}")
                return True
            else:
                logger.error("Elasticsearch连接失败：ping失败")
                return False
        except Exception as e:
            logger.error(f"Elasticsearch连接失败: {e}")
            return False
    
    def needs_batch_processing(self, sql):
        """判断是否为需要批量处理的SELECT语句"""
        sql = sql.strip().lower()
        # 只检查是否是SELECT语句，且不包含LIMIT
        return sql.startswith('select') and 'limit' not in sql

    def extract_table_name(self, sql):
        """从SQL语句中提取表名"""
        sql = sql.strip().lower()
        # 使用正则表达式提取from后面的表名
        match = re.search(r'from\s+([a-zA-Z0-9_\.]+)', sql)
        if match:
            return match.group(1)
        return None
    
    def check_table_size(self, database, table_name):
        """检查表的记录数是否超过阈值"""
        try:
            with self.mysql_conn.cursor() as cursor:
                self.mysql_conn.select_db(database)
                # 先检查表是否存在
                cursor.execute("SHOW TABLES LIKE %s", (table_name,))
                if not cursor.fetchone():
                    logger.warning(f"表 {table_name} 在数据库 {database} 中不存在")
                    return False
                
                # 获取表的大致行数
                # 注意：information_schema.TABLES中的TABLE_ROWS对InnoDB表只是估计值，可能与实际行数有很大差异
                # 原因：
                # 1. InnoDB存储引擎会使用采样方式统计表行数，不是精确计数
                # 2. 统计信息不会实时更新，需要执行ANALYZE TABLE命令刷新
                # 3. 对于小表和新建表，估计误差可能更大
                # 可以考虑使用SELECT COUNT(*) FROM table获取精确计数，但性能较低
                cursor.execute(f"SELECT TABLE_ROWS FROM information_schema.TABLES WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s", 
                              (database, table_name))
                result = cursor.fetchone()
                if result and 'TABLE_ROWS' in result:
                    table_rows = result['TABLE_ROWS']
                    logger.info(f"表 {table_name} 估计行数: {table_rows}")
                    return table_rows >= 300000  # 100万行阈值
                
                # 如果无法获取表行数，假设需要批量处理
                logger.warning(f"无法获取表 {table_name} 的行数信息，假设需要批量处理")
                return True
        except Exception as e:
            logger.error(f"检查表大小时出错: {e}")
            # 出错时默认进行批量处理，更安全
            return True

    def process_task(self, task):
        """处理单个同步任务"""
        stream_config = task.get('stream', {})
        jobs = task.get('jobs', [])
        
        database = stream_config.get('database')
        sql_template = stream_config.get('sql')
        pk_config = stream_config.get('pk', {})
        pk_field = pk_config.get('field', 'id')
        pk_type = pk_config.get('type', 'int')
        batch_size = stream_config.get('batch_size', 8000)  # 每批次处理的记录数
        
        logger.info(f"处理任务: 数据库={database}, 主键={pk_field}, 类型={pk_type}")
        logger.debug(f"任务详细配置: sql_template={sql_template}, batch_size={batch_size}")
        logger.debug(f"作业配置数量: {len(jobs)}")
        
        # 连接MySQL如果尚未连接
        if not self.mysql_conn or not self.mysql_conn.open:
            logger.debug("MySQL连接未建立或已关闭，尝试连接")
            if not self.connect_mysql():
                logger.error("无法连接MySQL，任务无法执行")
                return 0
        
        # 获取任务状态
        task_key = f"{database}_{pk_field}"
        last_id = self.sync_state.get(task_key, 0)
        max_id = last_id
        total_rows = 0
        
        logger.debug(f"任务状态: task_key={task_key}, last_id={last_id}")
        
        # 切换数据库
        self.mysql_conn.select_db(database)
        logger.debug(f"已切换到数据库: {database}")
        
        # 只有整数主键才进行分批处理
        if pk_type == 'int':
            logger.debug("检测到整数主键，可以进行分批处理")
            # 处理SELECT语句，检查是否需要转换为分批查询
            if self.needs_batch_processing(sql_template):
                logger.debug(f"SQL语句需要批处理: {sql_template}")
                table_name = self.extract_table_name(sql_template)
                logger.debug(f"提取的表名: {table_name}")
                if table_name:
                    # 检查表大小，只有超过100万行的表才进行分批处理
                    logger.debug(f"检查表 {table_name} 大小")
                    if self.check_table_size(database, table_name):
                        logger.info(f"检测到大型表查询，使用分批查询处理: {sql_template}")
                        logger.debug(f"开始分批处理表 {table_name}")
                        # 分批循环处理大表
                        batch_num = 0
                        while True:
                            batch_num += 1
                            logger.debug(f"处理第 {batch_num} 批数据, 起始ID: {last_id}")
                            # 替换SQL模板中的变量
                            current_sql = sql_template
                            if '{id}' in current_sql:
                                if pk_type == 'int':
                                    current_sql = current_sql.replace('{id}', str(last_id))
                                else:
                                    # 非整数类型，如果SQL中的占位符没有引号，需要添加引号
                                    if "'{id}'" in current_sql:
                                        current_sql = current_sql.replace('{id}', str(last_id))
                                    else:
                                        current_sql = current_sql.replace('{id}', str(last_id))
                                logger.debug(f"替换后的SQL: {current_sql}")
                            
                            # 如果没有主键条件，则转换为分批查询
                            if '{id}' not in sql_template:
                                logger.debug(f"SQL模板中没有ID占位符，转换为分批查询")
                                current_sql = self.convert_to_batch_query(sql_template, pk_field, batch_size, pk_type)
                                current_sql = current_sql.replace('{id}', str(last_id))
                                logger.debug(f"转换后的批处理SQL: {current_sql}")
                            
                            logger.debug(f"执行分批SQL: {current_sql}")
                            
                            # 执行当前批次查询
                            start_time = time.time()
                            with self.mysql_conn.cursor() as cursor:
                                cursor.execute(current_sql)
                                rows = cursor.fetchall()
                                
                                query_time = time.time() - start_time
                                logger.debug(f"查询执行时间: {query_time:.2f}秒")
                                
                                if not rows:
                                    logger.info(f"没有更多数据，批处理结束")
                                    break
                                
                                batch_rows = len(rows)
                                total_rows += batch_rows
                                logger.info(f"获取到第 {batch_num} 批数据，本批记录数: {batch_rows}，累计: {total_rows}")
                                
                                # 记录本批次ID范围
                                if batch_rows > 0 and pk_field in rows[0] and pk_field in rows[-1]:
                                    logger.info(f"本批次ID范围: {rows[0][pk_field]} - {rows[-1][pk_field]}")
                                
                                # 处理每个作业
                                for job_index, job in enumerate(jobs):
                                    logger.debug(f"处理第 {job_index+1}/{len(jobs)} 个作业")
                                    self.process_job(job, rows, pk_field)
                                
                                # 更新最大ID用于下一次查询
                                current_max_id = last_id
                                for row in rows:
                                    if pk_field in row and (pk_type == 'int' and int(row[pk_field]) > int(current_max_id)):
                                        current_max_id = row[pk_field]
                                
                                logger.debug(f"当前批次最大ID: {current_max_id}, 上一批次最大ID: {last_id}")
                                
                                if current_max_id == last_id:
                                    logger.warning(f"未找到比当前ID {last_id} 更大的记录，可能存在循环风险，终止批处理")
                                    break
                                
                                # 更新last_id为当前批次最大ID，用于下一次查询
                                last_id = current_max_id
                                if current_max_id > max_id:
                                    max_id = current_max_id
                                    logger.debug(f"更新总最大ID: {max_id}")
                                
                                # 如果本批次数据量小于batch_size，说明数据已经处理完毕
                                if batch_rows < batch_size:
                                    logger.info(f"数据不足一个批次({batch_rows} < {batch_size})，批处理结束")
                                    break
                        
                        # 批处理完成，更新同步状态
                        if max_id > self.sync_state.get(task_key, 0):
                            logger.info(f"更新同步状态: {task_key} = {max_id}")
                            self.sync_state[task_key] = max_id
                            self.save_sync_state()
                        
                        logger.info(f"分批处理任务完成, 共处理 {total_rows} 条记录")
                        return total_rows
                    else:
                        logger.info(f"表 {table_name} 记录数较少，不进行分批处理")
        
        # 对于非整数主键的表或记录数少的表，使用原来的处理方式
        logger.info(f"使用标准方式处理查询")
        logger.debug(f"标准处理: pk_type={pk_type}, last_id={last_id}")
        
        # 替换SQL模板中的变量
        if '{id}' in sql_template:
            if pk_type == 'int':
                sql = sql_template.replace('{id}', str(last_id))
            else:
                # 非整数类型，如果SQL中的占位符没有引号，需要添加引号
                if "'{id}'" in sql_template:
                    # 已有引号的情况
                    sql = sql_template.replace('{id}', str(last_id))
                else:
                    # 没有引号的情况，直接替换整个占位符（带引号）
                    sql = sql_template.replace('{id}', str(last_id))
            logger.debug(f"替换ID后的SQL: {sql}")
        else:
            sql = sql_template
            logger.debug(f"使用原始SQL(无ID占位符): {sql}")
        
        logger.debug(f"执行SQL: {sql}")
        
        # 执行查询
        start_time = time.time()
        with self.mysql_conn.cursor() as cursor:
            cursor.execute(sql)
            query_time = time.time() - start_time
            logger.debug(f"查询执行时间: {query_time:.2f}秒")
            
            # 分批处理结果
            batch_count = 0
            while True:
                batch_count += 1
                start_fetch = time.time()
                rows = cursor.fetchmany(self.bulk_size)
                fetch_time = time.time() - start_fetch
                
                if not rows:
                    logger.debug(f"没有更多数据，标准处理结束")
                    break
                
                total_rows += len(rows)
                logger.info(f"批次 {batch_count}：获取到 {len(rows)} 条记录，总计: {total_rows}")
                logger.debug(f"批次 {batch_count} 获取数据用时: {fetch_time:.2f}秒")
                
                # 记录本批次ID范围
                if len(rows) > 0 and pk_field in rows[0] and pk_field in rows[-1]:
                    logger.debug(f"本批次ID范围: {rows[0][pk_field]} - {rows[-1][pk_field]}")
                
                # 处理每个作业
                for job_index, job in enumerate(jobs):
                    logger.debug(f"处理第 {job_index+1}/{len(jobs)} 个作业")
                    self.process_job(job, rows, pk_field)
                
                # 更新最大ID
                for row in rows:
                    if pk_field in row and (pk_type == 'int' and int(row[pk_field]) > int(max_id) or 
                                            pk_type != 'int' and str(row[pk_field]) > str(max_id)):
                        max_id = row[pk_field]
            
            # 更新同步状态
            if max_id > self.sync_state.get(task_key, 0):
                logger.info(f"更新同步状态: {task_key} = {max_id}")
                self.sync_state[task_key] = max_id
                self.save_sync_state()
            
            logger.info(f"任务完成, 处理了 {total_rows} 条记录")
            return total_rows
    
    def convert_to_batch_query(self, sql, pk_field, batch_size, pk_type='int'):
        """将SQL语句转换为分批查询，使用SQLGlot处理复杂SQL结构"""
        logger.debug(f"转换SQL为分批查询: {sql}")
        
        # 如果SQL以分号结尾，去掉分号
        sql = sql.strip()
        if sql.endswith(';'):
            sql = sql[:-1]
        
        # 如果SQLGlot可用，使用它来解析和修改SQL
        if HAS_SQLGLOT:
            try:
                # 解析SQL，指定MySQL方言
                parsed = sqlglot.parse_one(sql, dialect='mysql')
                logger.debug(f"SQL解析成功: {parsed.sql(dialect='mysql')}")
                
                # 构建主键条件: pk_field > {id}
                if pk_type == 'int':
                    # 整数类型
                    pk_condition = sqlglot.exp.column(pk_field) > sqlglot.exp.Placeholder(this="{id}")
                else:
                    # 字符串类型，需要处理引号
                    pk_condition = sqlglot.exp.column(pk_field) > sqlglot.exp.Placeholder(this="{id}", quotes=True)
                
                logger.debug(f"构建的主键条件: {pk_condition.sql(dialect='mysql')}")
                
                # 添加或修改WHERE条件
                if parsed.args.get('where'):
                    # 已有WHERE条件，添加AND
                    parsed = parsed.where(sqlglot.exp.and_(parsed.args['where'], pk_condition), copy=False)
                    logger.debug(f"添加到现有WHERE条件")
                else:
                    # 没有WHERE条件，直接添加
                    parsed = parsed.where(pk_condition, copy=False)
                    logger.debug(f"添加新WHERE条件")
                
                # 添加ORDER BY (如果没有)
                if 'order' not in parsed.args or not parsed.args['order']:
                    parsed = parsed.order_by(pk_field, copy=False)
                    logger.debug(f"添加ORDER BY: {pk_field}")
                
                # 添加或替换LIMIT
                parsed = parsed.limit(batch_size, copy=False)
                logger.debug(f"设置LIMIT: {batch_size}")
                
                # 生成最终SQL
                result_sql = parsed.sql(dialect='mysql')
                logger.info(f"SQLGlot处理后的分批查询SQL: {result_sql}")
                
                # 确保占位符格式正确
                # 替换可能出现的参数化格式为MySQL支持的格式
                if pk_type == 'int':
                    # 处理数字类型
                    patterns = [':id', ':{id}', ':"{id}"', ':"id"', ":'id'", ":'{id}'"]
                    for pattern in patterns:
                        result_sql = result_sql.replace(pattern, '{id}')
                else:
                    # 处理字符串类型
                    patterns = [':"id"', ':"{id}"', ":'id'", ":'{id}'"]
                    targets = ['"{id}"', '"{id}"', "'{id}'", "'{id}'"]
                    for pattern, target in zip(patterns, targets):
                        result_sql = result_sql.replace(pattern, target)
                
                return result_sql
            except Exception as e:
                logger.warning(f"使用SQLGlot解析SQL失败: {e}")
                logger.debug(f"解析失败详情: {repr(e)}")
                # 如果SQLGlot解析失败，抛出异常不降级
                raise Exception('解析失败, 不降级')
        
        # 如果SQLGlot不可用或解析失败，使用简单的字符串处理作为后备方案
        logger.warning("使用简单字符串处理作为后备方案")
        sql_lower = sql.lower()
        
        # 准备ID占位符
        id_placeholder = "{id}" if pk_type == 'int' else "'{id}'"
        
        # 检查是否有WHERE子句
        if 'where' in sql_lower:
            # 找到WHERE和下一个子句
            where_pos = sql_lower.find('where')
            next_clause_pos = float('inf')
            for clause in ['group by', 'having', 'order by', 'limit']:
                pos = sql_lower.find(clause, where_pos)
                if pos > 0 and pos < next_clause_pos:
                    next_clause_pos = pos
            
            # 添加主键条件
            if next_clause_pos < float('inf'):
                # 在WHERE和下一个子句之间添加条件
                sql = f"{sql[:next_clause_pos]} AND {pk_field} > {id_placeholder} {sql[next_clause_pos:]}"
            else:
                # 直接在WHERE后添加条件
                sql = f"{sql} AND {pk_field} > {id_placeholder}"
        else:
            # 没有WHERE子句，添加一个
            # 找到FROM和下一个子句
            from_pos = sql_lower.find('from')
            next_clause_pos = float('inf')
            for clause in ['group by', 'having', 'order by', 'limit']:
                pos = sql_lower.find(clause, from_pos)
                if pos > 0 and pos < next_clause_pos:
                    next_clause_pos = pos
            
            # 添加WHERE子句
            if next_clause_pos < float('inf'):
                sql = f"{sql[:next_clause_pos]} WHERE {pk_field} > {id_placeholder} {sql[next_clause_pos:]}"
            else:
                sql = f"{sql} WHERE {pk_field} > {id_placeholder}"
        
        # 检查是否有ORDER BY子句
        if 'order by' not in sql_lower:
            # 找到LIMIT子句（如果有）
            limit_pos = sql_lower.find('limit')
            if limit_pos > 0:
                sql = f"{sql[:limit_pos]} ORDER BY {pk_field} {sql[limit_pos:]}"
            else:
                sql = f"{sql} ORDER BY {pk_field}"
        
        # 处理LIMIT子句
        if 'limit' in sql_lower:
            # 替换LIMIT值
            sql = re.sub(r'limit\s+\d+(?:\s*,\s*\d+)?', f'LIMIT {batch_size}', sql, flags=re.IGNORECASE)
        else:
            # 添加LIMIT
            sql = f"{sql} LIMIT {batch_size}"
        
        logger.info(f"字符串处理后的分批查询SQL: {sql}")
        return sql
    
    def process_job(self, job, rows, pk_field):
        """处理单个作业"""
        actions = job.get('actions', ['insert', 'update'])
        pipeline = job.get('pipeline', [])
        dest = job.get('dest', {})
        
        logger.debug(f"处理作业: actions={actions}, pipeline步骤数={len(pipeline)}")
        
        # 应用管道处理
        processed_rows = []
        start_time = time.time()
        for row in rows:
            processed_row = self.apply_pipeline(row, pipeline, pk_field)
            if processed_row:
                processed_rows.append(processed_row)
        
        pipeline_time = time.time() - start_time
        logger.debug(f"管道处理完成，处理 {len(rows)} 行数据用时: {pipeline_time:.2f}秒，有效行数: {len(processed_rows)}")
        
        # 发送到目标
        es_config = dest.get('es', {})
        if es_config:
            logger.debug(f"将数据发送到ES，配置: {es_config}")
            self.send_to_es(processed_rows, es_config, actions)
        else:
            logger.debug("没有配置ES目标，跳过发送")
    
    def apply_pipeline(self, row, pipeline, pk_field):
        """应用管道处理单条数据"""
        processed_row = row.copy()
        
        logger.debug(f"应用管道处理，步骤数: {len(pipeline)}")
        
        for step_index, step in enumerate(pipeline):
            logger.debug(f"执行管道步骤 {step_index+1}/{len(pipeline)}: {list(step.keys())}")
            
            # 设置ID
            if 'set_id' in step:
                id_field = step['set_id'].get('field', pk_field)
                processed_row['_id'] = processed_row.get(id_field)
                logger.debug(f"设置文档ID: {id_field} = {processed_row.get('_id')}")
            
            # 只保留指定字段
            if 'only_fields' in step:
                fields = step['only_fields'].get('fields', [])
                logger.debug(f"只保留字段: {fields}")
                
                filtered_row = {'_id': processed_row.get('_id')}
                for field in fields:
                    if field in processed_row:
                        filtered_row[field] = processed_row.get(field)
                    else:
                        logger.debug(f"字段 {field} 在原始数据中不存在")
                
                # 计算删除的字段数量
                removed_fields = len(processed_row) - len(filtered_row)
                logger.debug(f"保留 {len(filtered_row)} 个字段，删除 {removed_fields} 个字段")
                
                processed_row = filtered_row
            
            # 字段重命名
            if 'rename_fields' in step:
                rename_map = step['rename_fields'].get('mapping', {})
                logger.debug(f"字段重命名映射: {rename_map}")
                
                renamed_count = 0
                for old_name, new_name in rename_map.items():
                    if old_name in processed_row:
                        processed_row[new_name] = processed_row.pop(old_name)
                        renamed_count += 1
                    else:
                        logger.debug(f"要重命名的字段 {old_name} 在数据中不存在")
                
                logger.debug(f"成功重命名 {renamed_count}/{len(rename_map)} 个字段")
        
        logger.debug(f"管道处理完成，结果字段: {list(processed_row.keys())}")
        return processed_row
    
    def send_to_es(self, rows, es_config, actions):
        """发送数据到Elasticsearch"""
        index = es_config.get('index')
        doc_type = es_config.get('type', '_doc')  # ES 7.x 仍然需要传递类型但不再使用
        action_type = es_config.get('action', 'upsert')
        
        logger.info(f"发送 {len(rows)} 条数据到ES: index={index}, action={action_type}")
        
        # 准备批量操作
        bulk_actions = []
        skipped_rows = 0
        
        for row in rows:
            doc_id = row.pop('_id', None)
            
            if not doc_id:
                logger.warning(f"跳过没有ID的文档: {row}")
                skipped_rows += 1
                continue
            
            # 根据action_type决定操作类型
            if action_type == 'index' or 'insert' in actions:
                op_type = 'index'
                logger.debug(f"文档 {doc_id} 使用 index 操作")
            elif action_type == 'update' and 'update' in actions:
                op_type = 'update'
                row = {'doc': row, 'doc_as_upsert': True}
                logger.debug(f"文档 {doc_id} 使用 update 操作")
            elif action_type == 'upsert':
                op_type = 'update'
                row = {'doc': row, 'doc_as_upsert': True}
                logger.debug(f"文档 {doc_id} 使用 upsert 操作")
            else:
                logger.warning(f"不支持的操作类型: {action_type}，跳过文档 {doc_id}")
                skipped_rows += 1
                continue
            
            # ES 7.x不再需要在_type参数中指定文档类型
            action = {
                '_op_type': op_type,
                '_index': index,
                '_id': doc_id,
            }
            
            if op_type == 'update':
                action['doc'] = row['doc']
                action['doc_as_upsert'] = row['doc_as_upsert']
            else:
                action['_source'] = row
            
            bulk_actions.append(action)
        
        if skipped_rows > 0:
            logger.warning(f"准备批量操作时跳过了 {skipped_rows} 条无效记录")
        
        logger.debug(f"准备发送 {len(bulk_actions)} 条记录到ES索引 {index}")
        
        # 执行批量操作，增加错误处理
        if bulk_actions:
            start_time = time.time()
            try:
                success, errors = helpers.bulk(
                    self.es_client, 
                    bulk_actions, 
                    stats_only=True,
                    max_retries=3,
                    initial_backoff=2,
                    max_backoff=60
                )
                es_time = time.time() - start_time
                logger.info(f"ES批量操作完成: 成功={success}, 失败={errors}, 用时={es_time:.2f}秒")
                if errors:
                    logger.warning(f"有 {errors} 个文档操作失败")
            except Exception as e:
                es_time = time.time() - start_time
                logger.error(f"ES批量操作失败: {e}, 用时={es_time:.2f}秒")
                logger.debug(f"ES批量操作异常详情: {repr(e)}")
                
                # 失败后尝试分批重试
                if len(bulk_actions) > 1:
                    logger.info(f"尝试分批重新处理 {len(bulk_actions)} 个文档")
                    mid = len(bulk_actions) // 2
                    retry_start_time = time.time()
                    try:
                        # 处理前半部分
                        first_half_success, first_half_errors = helpers.bulk(
                            self.es_client, 
                            bulk_actions[:mid], 
                            stats_only=True,
                            max_retries=3
                        )
                        first_half_time = time.time() - retry_start_time
                        logger.info(f"前半部分 {mid} 个文档处理结果: 成功={first_half_success}, 失败={first_half_errors}, 用时={first_half_time:.2f}秒")
                    except Exception as e1:
                        first_half_time = time.time() - retry_start_time
                        logger.error(f"处理前半部分文档失败: {e1}, 用时={first_half_time:.2f}秒")
                        logger.debug(f"前半部分异常详情: {repr(e1)}")
                    
                    second_half_start_time = time.time()
                    try:
                        # 处理后半部分
                        second_half_success, second_half_errors = helpers.bulk(
                            self.es_client, 
                            bulk_actions[mid:], 
                            stats_only=True,
                            max_retries=3
                        )
                        second_half_time = time.time() - second_half_start_time
                        logger.info(f"后半部分 {len(bulk_actions) - mid} 个文档处理结果: 成功={second_half_success}, 失败={second_half_errors}, 用时={second_half_time:.2f}秒")
                    except Exception as e2:
                        second_half_time = time.time() - second_half_start_time
                        logger.error(f"处理后半部分文档失败: {e2}, 用时={second_half_time:.2f}秒")
                        logger.debug(f"后半部分异常详情: {repr(e2)}")
        else:
            logger.warning("没有有效的批量操作可执行")
    
    def run(self):
        """运行同步程序"""
        logger.info("开始同步程序")
        
        # 建立连接
        if not self.connect_mysql():
            return False
        
        if not self.connect_es():
            self.mysql_conn.close()
            return False
        
        try:
            # 处理所有任务
            for task in self.tasks:
                self.process_task(task)
            
            logger.info("所有任务处理完成")
            return True
            
        except Exception as e:
            logger.error(f"同步过程中发生错误: {e}", exc_info=True)
            return False
            
        finally:
            # 关闭连接
            if self.mysql_conn:
                self.mysql_conn.close()
            logger.info("同步程序结束")

def main():
    """主函数"""
    parser = argparse.ArgumentParser(description='MySQL到Elasticsearch数据同步工具')
    parser.add_argument('-c', '--config', help='配置文件路径')
    parser.add_argument('-l', '--log-level', choices=['DEBUG', 'INFO', 'WARNING', 'ERROR'], help='日志级别')
    args = parser.parse_args()
    
    # 配置文件优先级: 命令行参数 > 环境变量 > 默认值
    config_file = args.config  
    
    # 检查配置文件是否存在
    if not os.path.exists(config_file):
        # 在这里先临时设置日志，用于输出错误信息
        temp_logger = logging.getLogger('temp')
        handler = logging.StreamHandler()
        temp_logger.addHandler(handler)
        temp_logger.setLevel(logging.ERROR)
        temp_logger.error(f"配置文件不存在: {config_file}")
        temp_logger.error("请检查配置文件路径或设置CONFIG_FILE环境变量")
        return 1
    
    # 先加载配置文件，这样如果配置文件中有日志配置，就会先应用配置文件中的设置
    try:
        with open(config_file, 'r', encoding='utf-8') as f:
            if config_file.endswith('.py'):
                # 对于Python配置文件，执行它以应用其中的配置（包括日志配置）
                exec(f.read(), globals())
                print(f"已执行配置文件: {config_file}")
            else:
                # 对于YAML等其他类型配置文件的处理保持不变
                if config_file.endswith('.yaml') or config_file.endswith('.yml'):
                    import yaml
                    config = yaml.safe_load(f)
                else:
                    # 假设是其他格式，报错
                    print(f"不支持的配置文件格式: {config_file}")
                    return 1
    except Exception as e:
        print(f"加载配置文件时出错: {e}")
        return 1
    
    # 设置日志级别 - 在配置文件加载后执行，按优先级应用
    # 优先级顺序: 1. 命令行参数 2. 配置文件 3. 默认值(DEBUG)
    global logger
    # 获取配置文件中的日志级别配置
    config_log_level = globals().get('LEVEL', 'DEBUG')
    # 使用命令行参数的日志级别（如果提供），否则使用配置文件中的设置
    log_level = args.log_level if args.log_level else config_log_level
    # 获取配置文件中的日志格式配置
    log_format = globals().get('FORMAT', DEFAULT_LOG_FORMAT)
    
    logger = setup_logging(log_level, log_format)
    
    # 打印启动信息
    logger.info("=== MySQL到ES同步工具启动 ===")
    logger.info(f"配置文件: {config_file}")
    logger.info(f"数据目录: {DATA_DIR}")
    logger.info(f"日志目录: {LOGS_DIR}")
    logger.info(f"日志级别: {log_level}")
    logger.info(f"日志格式: {log_format}")
    logger.info("==========================")
    
    sync_tool = MySQLToESSync(config_file)
    sync_tool.run()

if __name__ == '__main__':
    main() 